package org.example
import org.apache.spark.sql.SparkSession
object fa {
  def main(args: Array[String]): Unit = {
    //SPARK运行环境
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("spark")
      .getOrCreate()

    val sc = spark.sparkContext
    //电影用户分析数据概况
    val filePath ="src/main/resources/"
    val moviesRDD = sc.textFile(filePath + "movies.dat")
    val occupationsRDD = sc.textFile(filePath + "occupations.dat")
    val ratingsRDD = sc.textFile(filePath + "ratings.dat")
    val usersRDD =sc.textFile(filePath + "users.dat")
    println("总电影数：" + moviesRDD.count())
    println("总用户数：" + usersRDD.count())
    println("评分条数：" + ratingsRDD.count())
    println("职业数：" + occupationsRDD.count())
    //1.用户分析：统计某用户评过分的电影数量并打印电影详情
    val userRating = ratingsRDD.map(_.split("::"))
      .map(user =>(user(1),user(0))) //转换为key（电影id）value（电影id）
      .filter(_._2.equals("2688")) //统计用户id为2688的影评数量
    println("用户id为2688的影评数量为：" + userRating.count())

    val movieInfo = moviesRDD.map(_.split("::"))
      .map(movie => (movie(0), (movie(1),movie(2))))
      .join(userRating)
      .map(item =>(item._1,item._2._2,item._2._1._1,item._2._1._2))
    movieInfo.take(10).foreach(println)

    //2.电影分析:找出平均得分最高的前10名电影（分数,电影Id）和观影人数最多的前10部电影(观影人数，电影Id)
    //平均得分最高的前10名电影
    val movieRatings = ratingsRDD.map(_.split("::"))
      .map { case Array(userId, movieId, rating, timestamp) => (movieId, rating.toDouble) }
      .aggregateByKey((0.0, 0))(
        (acc, value) => (acc._1 + value, acc._2 + 1),
        (acc1, acc2) => (acc1._1 + acc2._1, acc1._2 + acc2._2)
      )
      .mapValues { case (sum, count) => sum / count }
      .sortBy(_._2, ascending = false)
      .take(10)

    println("平均得分最高的前10名电影：")
    movieRatings.foreach { case (movieId, avgRating) =>
      println(s"电影ID: $movieId, 平均得分: $avgRating")
    }
    //观影人数最多的前10部电影
    val movieViewCounts = ratingsRDD.map(_.split("::")(1)) // 提取电影 ID
      .map((_, 1))
      .reduceByKey(_ + _)
      .sortBy(_._2, ascending = false)
      .take(10)

    println("观影人数最多的前10部电影：")
    movieViewCounts.foreach { case (movieId, viewCount) =>
      println(s"电影ID: $movieId, 观影人数: $viewCount")
    }


    sc.stop()
  }

}
